<!DOCTYPE HTML>
<!--
	Dimension by HTML5 UP
	html5up.net | @ajlkn
	Free for personal and commercial use under the CCA 3.0 license (html5up.net/license)
-->
<html>
 <head>
  <title>
   Dimension by HTML5 UP
  </title>
  <!-- <meta charset="utf-8" /> -->
  <!-- <meta name="viewport" content="width=device-width, initial-scale=1, user-scalable=no" /> -->
  <meta charset="utf-8"/>
  <meta content="width=device-width,initial-scale=1.0" name="viewport"/>
  <link href="../../assets/css/article.css" rel="stylesheet"/>
  <link href="https://cdn.bootcss.com/highlight.js/9.15.8/styles/github.min.css" rel="stylesheet"/>
  <noscript>
   <link href="../../assets/css/noscript.css" rel="stylesheet"/>
  </noscript>
 </head>
 <body>
  <div id="app">
  </div>
  <!-- built files will be auto injected -->
 </body>
 <body class="is-preload">
  <!-- Wrapper -->
  <div id="wrapper">
   <!-- Main -->
   <div id="main">
    <article id="article">
     <h1 id="java-kafka">
      Java Kafka 简单示例
     </h1>
     <hr/>
     <h2 id="_1">
      简介
     </h2>
     <p>
      Java kafka 简单代码示例
     </p>
     <h3 id="maven">
      maven依赖配置
     </h3>
     <div class="codehilite">
      <pre><span></span><code><span class="o">&lt;!--</span> <span class="n">kafka</span> <span class="o">--&gt;</span>
<span class="o">&lt;</span><span class="n">dependency</span><span class="o">&gt;</span>
    <span class="o">&lt;</span><span class="n">groupId</span><span class="o">&gt;</span><span class="n">org</span><span class="p">.</span><span class="na">apache</span><span class="p">.</span><span class="na">kafka</span><span class="o">&lt;/</span><span class="n">groupId</span><span class="o">&gt;</span>
    <span class="o">&lt;</span><span class="n">artifactId</span><span class="o">&gt;</span><span class="n">kafka</span><span class="o">-</span><span class="n">clients</span><span class="o">&lt;/</span><span class="n">artifactId</span><span class="o">&gt;</span>
    <span class="o">&lt;</span><span class="n">version</span><span class="o">&gt;</span><span class="mf">0.11.0.0</span><span class="o">&lt;/</span><span class="n">version</span><span class="o">&gt;</span>
<span class="o">&lt;/</span><span class="n">dependency</span><span class="o">&gt;</span>
</code></pre>
     </div>
     <h3 id="kakfa">
      kakfa生产和消费者生成
     </h3>
     <div class="codehilite">
      <pre><span></span><code><span class="kn">import</span> <span class="nn">org.apache.kafka.clients.consumer.ConsumerRecord</span><span class="p">;</span>
<span class="kn">import</span> <span class="nn">org.apache.kafka.clients.consumer.ConsumerRecords</span><span class="p">;</span>
<span class="kn">import</span> <span class="nn">org.apache.kafka.clients.consumer.KafkaConsumer</span><span class="p">;</span>
<span class="kn">import</span> <span class="nn">org.apache.kafka.clients.producer.KafkaProducer</span><span class="p">;</span>
<span class="kn">import</span> <span class="nn">org.apache.kafka.clients.producer.ProducerRecord</span><span class="p">;</span>

<span class="kn">import</span> <span class="nn">java.util.*</span><span class="p">;</span>

<span class="cm">/**</span>
<span class="cm"> * @author lw1243925457</span>
<span class="cm"> */</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">KafkaUtil</span> <span class="p">{</span>

    <span class="kd">public</span> <span class="kd">static</span> <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="nf">createConsumer</span><span class="p">(</span><span class="n">String</span> <span class="n">servers</span><span class="p">,</span> <span class="n">String</span> <span class="n">topic</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">Properties</span> <span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="p">();</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"bootstrap.servers"</span><span class="p">,</span> <span class="n">servers</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"group.id"</span><span class="p">,</span> <span class="s">"group-1"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"enable.auto.commit"</span><span class="p">,</span> <span class="s">"false"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"auto.commit.interval.ms"</span><span class="p">,</span> <span class="s">"1000"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"auto.offset.reset"</span><span class="p">,</span> <span class="s">"earliest"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"session.timeout.ms"</span><span class="p">,</span> <span class="s">"30000"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"key.deserializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringDeserializer"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"value.deserializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringDeserializer"</span><span class="p">);</span>

        <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">kafkaConsumer</span> <span class="o">=</span> <span class="k">new</span> <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span><span class="p">(</span><span class="n">properties</span><span class="p">);</span>
        <span class="n">kafkaConsumer</span><span class="p">.</span><span class="na">subscribe</span><span class="p">(</span><span class="n">Arrays</span><span class="p">.</span><span class="na">asList</span><span class="p">(</span><span class="n">topic</span><span class="p">));</span>
        <span class="k">return</span> <span class="n">kafkaConsumer</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="kd">public</span> <span class="kd">static</span> <span class="kt">void</span> <span class="nf">readMessage</span><span class="p">(</span><span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">kafkaConsumer</span><span class="p">,</span> <span class="kt">int</span> <span class="n">timeout</span><span class="p">)</span> <span class="p">{</span>
        <span class="k">while</span> <span class="p">(</span><span class="kc">true</span><span class="p">)</span> <span class="p">{</span>
            <span class="n">ConsumerRecords</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">records</span> <span class="o">=</span> <span class="n">kafkaConsumer</span><span class="p">.</span><span class="na">poll</span><span class="p">(</span><span class="n">timeout</span><span class="p">);</span>
            <span class="k">for</span> <span class="p">(</span><span class="n">ConsumerRecord</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">record</span> <span class="p">:</span> <span class="n">records</span><span class="p">)</span> <span class="p">{</span>
                <span class="n">String</span> <span class="n">value</span> <span class="o">=</span> <span class="n">record</span><span class="p">.</span><span class="na">value</span><span class="p">();</span>
                <span class="n">kafkaConsumer</span><span class="p">.</span><span class="na">commitAsync</span><span class="p">();</span>
                <span class="n">System</span><span class="p">.</span><span class="na">out</span><span class="p">.</span><span class="na">println</span><span class="p">(</span><span class="n">value</span><span class="p">);</span>
            <span class="p">}</span>
        <span class="p">}</span>
    <span class="p">}</span>

    <span class="kd">public</span> <span class="kd">static</span> <span class="n">KafkaProducer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="nf">createProducer</span><span class="p">(</span><span class="n">String</span> <span class="n">servers</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">Properties</span> <span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="p">();</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"bootstrap.servers"</span><span class="p">,</span> <span class="n">servers</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"acks"</span><span class="p">,</span> <span class="s">"all"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"retries"</span><span class="p">,</span> <span class="mi">0</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"batch.size"</span><span class="p">,</span> <span class="mi">16384</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"linger.ms"</span><span class="p">,</span> <span class="mi">1</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"buffer.memory"</span><span class="p">,</span> <span class="mi">33554432</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"key.serializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringSerializer"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"value.serializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringSerializer"</span><span class="p">);</span>
        <span class="k">return</span> <span class="k">new</span> <span class="n">KafkaProducer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span><span class="p">(</span><span class="n">properties</span><span class="p">);</span>
    <span class="p">}</span>

    <span class="kd">public</span> <span class="kd">static</span> <span class="kt">void</span> <span class="nf">send</span><span class="p">(</span><span class="n">KafkaProducer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">producer</span><span class="p">,</span> <span class="n">String</span> <span class="n">topic</span><span class="p">,</span> <span class="n">String</span> <span class="n">message</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">producer</span><span class="p">.</span><span class="na">send</span><span class="p">(</span><span class="k">new</span> <span class="n">ProducerRecord</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span><span class="p">(</span><span class="n">topic</span><span class="p">,</span> <span class="n">message</span><span class="p">));</span>
    <span class="p">}</span>
<span class="p">}</span>
</code></pre>
     </div>
     <h3 id="_2">
      运行
     </h3>
     <div class="codehilite">
      <pre><span></span><code><span class="kd">public</span> <span class="kd">class</span> <span class="nc">Main</span> <span class="p">{</span>

    <span class="kd">public</span> <span class="kd">static</span> <span class="kt">void</span> <span class="nf">main</span><span class="p">(</span><span class="n">String</span><span class="o">[]</span> <span class="n">args</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">String</span> <span class="n">servers</span> <span class="o">=</span> <span class="s">"localhost:9092,localhost:9093,localhost:9094"</span><span class="p">;</span>
        <span class="n">String</span> <span class="n">topic</span> <span class="o">=</span> <span class="s">"TestTopic"</span><span class="p">;</span>
        <span class="n">String</span> <span class="n">message</span> <span class="o">=</span> <span class="s">"test"</span><span class="p">;</span>

        <span class="n">KafkaProducer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">producer</span> <span class="o">=</span> <span class="n">KafkaUtil</span><span class="p">.</span><span class="na">createProducer</span><span class="p">(</span><span class="n">servers</span><span class="p">);</span>
        <span class="n">KafkaUtil</span><span class="p">.</span><span class="na">send</span><span class="p">(</span><span class="n">producer</span><span class="p">,</span> <span class="n">topic</span><span class="p">,</span> <span class="n">message</span><span class="p">);</span>

        <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">consumer</span> <span class="o">=</span> <span class="n">KafkaUtil</span><span class="p">.</span><span class="na">createConsumer</span><span class="p">(</span><span class="n">servers</span><span class="p">,</span> <span class="n">topic</span><span class="p">);</span>
        <span class="n">KafkaUtil</span><span class="p">.</span><span class="na">readMessage</span><span class="p">(</span><span class="n">consumer</span><span class="p">,</span> <span class="mi">100</span><span class="p">);</span>
    <span class="p">}</span>
<span class="p">}</span>
</code></pre>
     </div>
     <h2 id="_3">
      使用心得
     </h2>
     <h3 id="_4">
      总是读取最老的消息
     </h3>
     <p>
      可能是group-id的问题，新起一个group-id名称
     </p>
     <ul>
      <li>
       earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      </li>
      <li>
       latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      </li>
      <li>
       none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      </li>
     </ul>
     <h2 id="_5">
      参考链接
     </h2>
     <ul>
      <li>
       <a href="https://blog.csdn.net/beExcellentOne/article/details/53641953">
        java 实现kafka消息生产者和消费者
       </a>
      </li>
      <li>
       <a href="https://segmentfault.com/a/1190000015886487">
        kafka（三）—Kafka的Java代码示例和配置说明
       </a>
      </li>
      <li>
       <a href="https://blog.csdn.net/u011669700/article/details/80053313">
        Kafka - 偏移量提交
       </a>
      </li>
      <li>
       <a href="http://www.dengshenyu.com/%E5%88%86%E5%B8%83%E5%BC%8F%E7%B3%BB%E7%BB%9F/2017/11/14/kafka-consumer.html">
        Kafka系列（四）Kafka消费者：从Kafka中读取数据
       </a>
      </li>
      <li>
       <a href="https://blog.csdn.net/lishuangzhe7047/article/details/74530417">
        Kafka auto.offset.reset值详解
       </a>
      </li>
     </ul>
    </article>
   </div>
   <!-- Footer -->
   <footer id="footer">
    <p class="copyright">
     © Untitled. Design:
     <a href="https://html5up.net">
      HTML5 UP
     </a>
     .
    </p>
   </footer>
  </div>
  <!-- BG -->
  <div id="bg">
  </div>
  <!-- Scripts -->
  <script src="../assets/js/jquery.min.js">
  </script>
  <script src="../assets/js/browser.min.js">
  </script>
  <script src="../assets/js/breakpoints.min.js">
  </script>
  <script src="../assets/js/util.js">
  </script>
  <script src="../assets/js/main.js">
  </script>
 </body>
</html>
